package com.hzw.saas.service.rss.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.digest.DigestUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SimplePropertyPreFilter;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.IService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hzw.saas.api.rss.*;
import com.hzw.saas.api.rss.bo.JournalArticleBO;
import com.hzw.saas.api.rss.bo.JournalArticleNoticeBO;
import com.hzw.saas.api.rss.bo.JournalBO;
import com.hzw.saas.api.rss.bo.JournalUserSubBO;
import com.hzw.saas.api.rss.query.JournalSubQuery;
import com.hzw.saas.common.config.annotation.RedisProcessLock;
import com.hzw.saas.common.config.util.AssertUtil;
import com.hzw.saas.common.config.util.ExcelUtils;
import com.hzw.saas.common.config.util.PageParam;
import com.hzw.saas.common.config.util.PageUtils;
import com.hzw.saas.common.util.MapperUtil;
import com.hzw.saas.common.util.SnowFlakeUtil;
import com.hzw.saas.service.rss.dao.RssElasticRepository;
import com.hzw.saas.service.rss.mapper.JournalMapper;
import com.hzw.saas.service.rss.model.Journal;
import com.hzw.saas.service.rss.model.JournalArticle;
import com.hzw.saas.service.rss.model.JournalArticleNotice;
import com.hzw.saas.service.rss.parse.RSSParser;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * <p>
 * 期刊站点 服务实现类
 * </p>
 *
 * @author sonam
 * @since 2021-06-30
 */
@Slf4j
@RequiredArgsConstructor
@Service("journalService")
public class JournalServiceImpl extends ServiceImpl<JournalMapper, Journal> implements IJournalService, IService<Journal> {

    @Autowired
    private JournalMapper journalMapper;

    static SimplePropertyPreFilter filter = new SimplePropertyPreFilter();

    static {
        filter.getExcludes().add("source");
        filter.getExcludes().add("digest");
        filter.getExcludes().add("linkUrl");
        filter.getExcludes().add("subject");
        filter.getExcludes().add("creatorId");
        filter.getExcludes().add("createTime");
        filter.getExcludes().add("updateTime");
        filter.getExcludes().add("pubDate");
        filter.getExcludes().add("editorId");
        filter.getExcludes().add("authors");
        filter.getExcludes().add("pid");
        filter.getExcludes().add("esSync");
        filter.getExcludes().add("image_title");
    }


    @Resource(name = "journalArticleService")
    private IJournalArticleService journalArticleService;

    @Resource(name = "journalUserSubService")
    private IJournalUserSubService journalUserSubService;

    @Resource(name = "journalCustomService")
    private IJournalCustomService journalCustomService;

    private final RssElasticRepository rssElasticRepository;

    @Resource(name = "rssEsService")
    private IRssEsService rssEsService;

    @Override
    public JournalBO getById(String pid) {
        if (StrUtil.isBlank(pid)) return null;
        return MapperUtil.nf().map(super.getById(pid), JournalBO.class);
    }

    @Override
    public List<JournalBO> query(JournalBO journalBO) {
        if (Objects.isNull(journalBO)) return MapperUtil.nf().mapAsList(this.list(), JournalBO.class);
        Journal journal = MapperUtil.nf().map(journalBO, Journal.class);
        List<Journal> journals = this.list(Wrappers.<Journal>query().setEntity(journal));
        return MapperUtil.nf().mapAsList(journals, JournalBO.class);
    }

    @Override
    public void update(JournalBO journalBO) {
        if (journalBO == null) return;
        Journal journal = MapperUtil.nf().map(journalBO, Journal.class);
        this.updateById(journal);
    }

    @Override
    public void delete(JournalBO journalBO) {
        if (journalBO == null) return;
        this.remove(Wrappers.<Journal>query().setEntity(MapperUtil.nf().map(journalBO, Journal.class)));
    }

    @Override
    public void delete(String pid) {
        if (StrUtil.isBlank(pid)) return;
        this.removeById(pid);
    }

    @Override
    public void saveBatch(List<JournalBO> journalBOS) {
        if (CollectionUtil.isEmpty(journalBOS)) return;
        journalBOS.forEach(journalBO -> {
            String pid = journalBO.getPid();
            if (StrUtil.isBlank(pid)) {
                journalBO.setPid(SnowFlakeUtil.nextIdStr());
                //todo 这里写死类型，后续跟前端调
            }
            journalBO.setJournalType(2);
        });
        this.saveBatch(MapperUtil.nf().mapAsList(journalBOS, Journal.class));
    }

    @Override
    public IPage<JournalBO> pagePublicByCatalogIds(PageParam pageParam, List<String> catalogIds) {
        if (pageParam == null) {
            pageParam = new PageParam();
        }
        Page<Journal> journalPage = this.page(pageParam.convert(), Wrappers.<Journal>lambdaQuery()
            .in(CollectionUtil.isNotEmpty(catalogIds), Journal::getCatalogId, catalogIds));
        return PageUtils.convert(journalPage, JournalBO.class);
    }


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void importRss(String filePath) throws IOException {
        List<JournalBO> journalBOS = ExcelUtils.importExcel(filePath, 1, 1, JournalBO.class);
        journalBOS.forEach(journalBO -> journalBO.setPid(SnowFlakeUtil.nextIdStr()));
        List<Journal> journals = MapperUtil.nf().mapAsList(journalBOS, Journal.class);
        saveBatch(journals);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    @RedisProcessLock(lockName = "UpdateMySubscribe", key = "#journalSubQuery.userId+':'+#journalSubQuery.rssId")
    public void subJournalRss(@Validated JournalSubQuery journalSubQuery) {
        AssertUtil.assertThrow("订阅失败，期刊RSS不存在", journalCustomService.findCountLeZero(journalSubQuery.getRssId()));
        sub(journalSubQuery);
    }

    @Override
    public void sub(JournalSubQuery journalSubQuery) {
        JournalUserSubBO condition = new JournalUserSubBO()
            .setJournalId(journalSubQuery.getRssId())
            .setUserId(journalSubQuery.getUserId());
        //取消订阅
        if (journalSubQuery.getSub() == 0) {
            journalUserSubService.delete(condition);
            return;
        }
        AssertUtil.assertThrow("订阅失败，参数错误", journalSubQuery.getSub() != 1);

        List<JournalUserSubBO> journalUserSubBOS = journalUserSubService.query(condition);
        if (CollectionUtil.isNotEmpty(journalUserSubBOS)) {
            // 已订阅直接返回
            return;
        }
        //订阅
        JournalUserSubBO journalUserSubBO = new JournalUserSubBO();
        journalUserSubBO.setUserId(journalSubQuery.getUserId());
        journalUserSubBO.setJournalId(journalSubQuery.getRssId());
        journalUserSubBOS.add(journalUserSubBO);
        journalUserSubService.addJournalUserSub(journalUserSubBOS);
    }


    @Override
    public IPage<JournalBO> findMySubscribe(PageParam pageParam, String userId) {
        if (pageParam == null) {
            pageParam = new PageParam();
        }

        List<JournalUserSubBO> journalUserSubByUserId = journalUserSubService.findJournalUserSubByUserId(userId);
        if (CollectionUtil.isEmpty(journalUserSubByUserId)) {
            return new Page<>();
        }
        List<String> collect = journalUserSubByUserId.stream().map(JournalUserSubBO::getJournalId).collect(Collectors.toList());

        Page customPage = journalCustomService.findCustomPage(pageParam, collect);

        return PageUtils.convert(customPage, JournalBO.class);
    }

    @Override
    public List<JournalBO> findCatalogByIds(List<String> ids) {
        List<Journal> list = this.lambdaQuery().in(Journal::getCatalogId, ids).list();
        return MapperUtil.nt().mapAsList(list, JournalBO.class);
    }


    @Override
    public void parseJob(String journalId) {
        saveJournal(this.lambdaQuery().eq(StrUtil.isNotBlank(journalId), Journal::getPid, journalId).list());
    }

    private void saveJournal(List<Journal> journals) {
        if (CollectionUtil.isEmpty(journals)) {
            return;
        }
        journals.forEach(journal -> {
            log.info("开始提取期刊 {}", journal.getPid());
            String content;
            try {
                content = HttpUtil.get(journal.getRssUrl(), 60000);
            } catch (Exception e) {
                log.error(String.format("journal rss [%s] connect failed", journal.getRssUrl()), e);
                return;
            }

            List<JournalArticleBO> journalArticleBOS;
            try {
                journalArticleBOS = RSSParser.parse(content);
            } catch (Exception e) {
                log.error(String.format("journal rss [%s] parse err", journal.getRssUrl()), e);
                return;
            }
            List<JournalArticleBO> articleBOS = new ArrayList<>();
            try {
                if (CollectionUtil.isEmpty(journalArticleBOS)) {
                    return;
                }
                journalArticleBOS.forEach(journalArticleBO -> {
                    if (journalArticleBO == null) {
                        return;
                    }
                    journalArticleBO.setRssId(journal.getPid());
                    String json = JSONObject.toJSONString(journalArticleBO, filter);
                    String digest = DigestUtil.sha256Hex(json);
                    journalArticleBO.setDigest(digest);
                    if (StrUtil.isNotBlank(digest)) {
                        Integer repeatDigestCount = journalArticleService.findRepeatDigestCount(journalArticleBO.getDigest());
                        if (repeatDigestCount == 0) {
                            journalArticleBO.setPid(SnowFlakeUtil.nextIdStr());
                            articleBOS.add(journalArticleBO);
                        }
                    }

                });
                //批量存储
                journalArticleService.saveBatch(articleBOS);
                journal.setSyncTime(new Date());
                this.updateById(journal);
                log.info("rss-journal-article new size:{}, old size: {}", articleBOS.size(), journalArticleBOS.size());
            } catch (Exception e) {
                log.error(String.format("journal rss [%s] store err", journal.getRssUrl()), e);
            }
        });
    }

    @Override
    public String articleDigest(JournalArticleBO journalArticleBO) {
        String json = JSONObject.toJSONString(journalArticleBO, filter);
        return DigestUtil.sha256Hex(json);
    }

    @Override
    public void refreshAll() {
        rssElasticRepository.deleteAll();
        refreshES();
    }

    @Override
    @Transactional
    public void refreshES() {
        JournalArticleServiceImpl journalArticleService = (JournalArticleServiceImpl) this.journalArticleService;
        // 查询未同步的RSS期刊
        List<JournalArticle> journalArticles = journalArticleService.lambdaQuery().eq(JournalArticle::getEsSync, 0).list();
        writeES(MapperUtil.nf().mapAsList(journalArticles, JournalArticleBO.class));
    }

    private void writeES(List<JournalArticleBO> journalArticleBOS) {
        if (CollectionUtil.isEmpty(journalArticleBOS)) {
            log.info("没有需要同步的RSS期刊数据");
            return ;
        }
        // 处理作者以及日期时间 +8 hours
        journalArticleBOS.forEach(journalArticleBO -> {
            String author = journalArticleBO.getAuthor();
            if (StrUtil.isBlank(author)) {
                return;
            }
            String[] authors = author.split(StrUtil.COMMA);
            List<String> strings = new ArrayList<>();
            for(String s1 : authors) {
                strings.add(s1.trim());
            }
            journalArticleBO.setAuthors(strings);
            if (journalArticleBO.getPubDate() != null) {
                journalArticleBO.setPubDate(DateUtil.offset(journalArticleBO.getPubDate(), DateField.HOUR, 8));
            }
        });
        List<JournalArticleNoticeBO> journalArticleNotices = MapperUtil.nf().mapAsList(journalArticleBOS, JournalArticleNoticeBO.class);
        List<String> rssIds = journalArticleNotices.stream().map(JournalArticleNoticeBO::getRssId).distinct().collect(Collectors.toList());
        //根据rssId查询es其他属性，用于检索
        List<Journal> queryJournals = this.lambdaQuery().in(Journal::getPid, rssIds).list();
        if (CollectionUtil.isNotEmpty(queryJournals)) {
            journalArticleNotices.forEach(journalArticleNotice -> {
                queryJournals.forEach(queryJournal -> {
                    if (journalArticleNotice.getRssId().equals(queryJournal.getPid())) {
                        journalArticleNotice.setJournalIssn(queryJournal.getJournalIssn());
                        journalArticleNotice.setJournalName(queryJournal.getJournalName());
                        journalArticleNotice.setCatalogId(queryJournal.getCatalogId());
                    }
                });
            });
        }
        //es分段提交
        List<JournalArticleNotice> journalArticleNoticeList = MapperUtil.nf().mapAsList(journalArticleNotices, JournalArticleNotice.class);
        int batchSize = 300;
        Iterator<JournalArticleNotice> iterator = journalArticleNoticeList.iterator();
        List<JournalArticleNotice> tempJournalArticleList = new ArrayList<>();
        while(iterator.hasNext()) {
            JournalArticleNotice journalArticleNotice = iterator.next();
            tempJournalArticleList.add(journalArticleNotice);
            if(tempJournalArticleList.size() >= batchSize) {
                rssElasticRepository.saveAll(tempJournalArticleList);
                tempJournalArticleList.clear();
            }
        }
        // 最后在保存一次
        if(CollectionUtil.isNotEmpty(tempJournalArticleList)) {
            rssElasticRepository.saveAll(tempJournalArticleList);
        }
        // 设置RSS期刊状态为已同步
        JournalArticleServiceImpl journalArticleService = (JournalArticleServiceImpl) this.journalArticleService;
        List<JournalArticle> journalArticles = new ArrayList<>();
        journalArticleNoticeList.forEach(journalArticleNotice -> {
            JournalArticle journalArticle = new JournalArticle();
            journalArticle.setEsSync(1); // 设置已同步
            journalArticle.setPid(journalArticleNotice.getPid()); // 设置已同步
            journalArticles.add(journalArticle);
        });
        // 更新RSS期刊，设置状态为已同步
        if(CollectionUtil.isNotEmpty(journalArticles)) {
            journalArticleService.updateBatchById(journalArticles, 1000);
        }
    }

    @Override
    public List<String> findUnSave() {
        List<Journal> list = this.lambdaQuery().list();
        if (CollectionUtil.isNotEmpty(list)) {
            ArrayList<String> arrayList = new ArrayList<>();
            //所有需要被查询的文章id
            List<String> collect = list.stream().map(Journal::getPid).collect(Collectors.toList());
            collect.forEach(s -> {
                ArrayList<String> stringArrayList = new ArrayList<>();
                stringArrayList.add(s);
                List<JournalArticleBO> byJournalIds = journalArticleService.findByJournalIds(stringArrayList);
                if (CollectionUtil.isNotEmpty(byJournalIds)) {
                    List<JournalArticleNoticeBO> byJournalId = rssEsService.findByJournalId(s, byJournalIds.size());
                    //数据库中的主键
                    List<String> collect1 = byJournalIds.stream().map(JournalArticleBO::getPid).collect(Collectors.toList());
                    if (CollectionUtil.isEmpty(byJournalId)) {
                        arrayList.addAll(collect1);
                    }
                    //es中的主键
                    List<String> strings = byJournalId.stream().map(JournalArticleNoticeBO::getPid).collect(Collectors.toList());
                    collect1.removeAll(strings);
                    if (collect1.size() != 0) {
                        arrayList.addAll(collect1);
                    }
                }
            });
            return arrayList;
        }
        return Collections.emptyList();
    }

    @Override
    public List<JournalBO> findByIds(List<String> collect) {
        if (CollectionUtil.isNotEmpty(collect)) {
            List<Journal> list = this.lambdaQuery().in(Journal::getPid, collect).list();
            return MapperUtil.nf().mapAsList(list, JournalBO.class);
        }
        return null;
    }

    @Override
    public List<JournalBO> findByCatalogIds(List<String> collect) {
        if (CollectionUtil.isNotEmpty(collect)) {
            List<Journal> list = this.lambdaQuery().in(Journal::getCatalogId, collect).list();
            return MapperUtil.nf().mapAsList(list, JournalBO.class);
        }
        return null;
    }

    @Override
    public List<JournalBO> findRepeatUrl(List<JournalBO> collect) {
        if (CollectionUtil.isNotEmpty(collect)) {
            List<String> collect1 = collect.stream().map(JournalBO::getRssUrl).collect(Collectors.toList());
            List<Journal> list = this.lambdaQuery().in(Journal::getRssUrl, collect1).list();
            return MapperUtil.nf().mapAsList(list, JournalBO.class);

        }
        return null;
    }


}
